package com.oracle.violet.first.area

import com.oracle.violet.first.bean.ProArea
import com.oracle.violet.first.config.ConfigHandler
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}

object AreaDistCore {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("地域分布")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)

    //读取文件
    val dataFrame: DataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)
    import sQLContext.implicits._

    //提取维度:省市
    dataFrame.map(f = str => {
      val pname = str.getAs[String] ("provincename")
      val cname = str.getAs[String]("cityname")

      //提取需求需要的相应字段
      val reqmode = str.getAs[Int]("requestmode")
      val pronode = str.getAs[Int]("processnode")
      val iseffect = str.getAs[Int]("iseffective")
      val isbiling = str.getAs[Int]("isbilling")
      val isbid = str.getAs[Int]("isbid")
      val iswin = str.getAs[Int]("iswin")
      val adorderid = str.getAs[Int]("adorderid")


      val adRawReq=if(reqmode==1 && pronode>=1) 1 else 0
      val adEffReq=if(reqmode==1 && pronode>=2) 1 else 0
      val adReq=if(reqmode==1 && pronode==3) 1 else 0

      val adRtbsReq=if(iseffect == 1 && isbiling == 1 && isbid == 1 && adorderid != 0) 1 else 0
      val adRtbReq = if (iseffect == 1 && isbiling == 1 && iswin==1) {
        val winprice = str.getAs[Double]("winprice")
        val adpayment = str.getAs[Double]("adpayment")
        (1,winprice/1000,adpayment/1000)
      }else (0,0d,0d)
      val adshow=if(reqmode==2 && iseffect==1) 1 else 0
      val adClick=if(reqmode==3 && iseffect==1) 1 else 0

      //组装成元组
      ((pname,cname),List(adRawReq,adEffReq,adReq,adRtbsReq,adRtbReq._1,adshow,adClick,adRtbReq._2,adRtbReq._3))
    }).reduceByKey((a,b)=>a.zip(b).map(t=>t._1+t._2))
    //zip表示取到数组的列
    //封装
      .map(tp=>ProArea(
      tp._1._1,
      tp._1._2,
      tp._2(0).toInt,
      tp._2(1).toInt,
      tp._2(2).toInt,
      tp._2(3).toInt,
      tp._2(4).toInt,
      tp._2(5).toInt,
      tp._2(6).toInt,
      tp._2(7),
      tp._2(8))).toDF().write.mode(SaveMode.Overwrite).jdbc(ConfigHandler.url, "orc_AreaDistCore_province", ConfigHandler.props)


   sc.stop()
  }

}
